package com.dahantc;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;
import java.util.Map;

@Slf4j
public class KafkaUtils {

    public static Map<String, Object> buildConfig(boolean isConsumer) {
        Map<String, Object> configs = new HashMap<>();
        // 设置连接Kafka的初始连接⽤到的服务器地址
        // 如果是集群，则可以通过此初始连接发现集群中的其他broker
        configs.put("bootstrap.servers", KafkaApp.KAFKA_URL);
        configs.put("acks", "-1");
        configs.put("session.timeout.ms", 600000);

        if ("T".equals(KafkaApp.IS_NEED_PWD)){
            configs.put("security.protocol", "SASL_PLAINTEXT");

            String jassc;
            switch (KafkaApp.MECHANISM) {
                case "PLAIN":
                    configs.put("sasl.mechanism", "PLAIN");
                    jassc = "org.apache.kafka.common.security.plain.PlainLoginModule required\n"
                            + "username=\"" + KafkaApp.KAFKA_USER + "\"\n"
                            + "password=\"" + KafkaApp.KAFKA_PASSWORD + "\";";
                    configs.put("sasl.jaas.config", jassc);
                    break;
                case "SCRAM-SHA-512":
                    configs.put("sasl.mechanism", "SCRAM-SHA-512");
                    jassc = "org.apache.kafka.common.security.scram.ScramLoginModule required\n"
                            + "username=\"" + KafkaApp.KAFKA_USER + "\"\n"
                            + "password=\"" + KafkaApp.KAFKA_PASSWORD + "\";";
                    configs.put("sasl.jaas.config", jassc);
                    break;
                default:
                    log.error("\n\n=========  System start failed, Error mechanism :  . ===========\n\n" + KafkaApp.MECHANISM);
                    System.exit(0);
            }

        }
        if (isConsumer) {
            configs.put("max.poll.interval.ms", 600000);
            // 设置key的序列化器
            configs.put("key.deserializer", StringDeserializer.class);
            // 设置value的序列化器
            configs.put("value.deserializer", StringDeserializer.class);
            configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
            configs.put("enable.auto.commit", "false");
            configs.put("auto.commit.interval.ms", "3000");
            configs.put("max.partition.fetch.bytes", "10240");//每次拉取的消息字节数，10K，每次取回20条左右
        } else {
            configs.put("producer.acks", "1");
            configs.put("producer.retries", 3);
            // 设置key的序列化器
            configs.put("key.serializer", StringSerializer.class);
            // 设置value的序列化器
            configs.put("value.serializer", StringSerializer.class);
            // configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transaction_id_0");
            // configs.put("enable.idempotence",true); // 设置幂等性
        }
        return configs;
    }

}
